
#define SPDLOG_HEADER_ONLY
#define FMT_HEADER_ONLY
#include <spdlog/spdlog.h>
#include <spdlog/async.h>
#include <spdlog/sinks/stdout_color_sinks.h>
#include "spdlog/sinks/rotating_file_sink.h"

#include "hsocket.h"

#include "latch.hpp"
#include "lua_network.hpp"

static latch latch_;

static void on_mqtt(mqtt_client_t* cli, int type)
{
    MQTT* mqtt = (MQTT*)cli->userdata;

    switch (type)
    {
    case MQTT_TYPE_CONNACK:
    {
        auto       peeraddr         = hio_localaddr(cli->io);
        char       localaddr[128] = {0};
        sockaddr_u u{*peeraddr};

        mqtt_lua_call(mqtt->get_connected_fn(), fmt::format("{}:{}({})", cli->host, cli->port, cli->client_id).c_str(),
                      fmt::format("{}:{}", sockaddr_ip(&u, localaddr, sizeof(localaddr)), sockaddr_port(&u)).c_str());

        latch_.count_down();
    }
    break;
    case MQTT_TYPE_PUBLISH:
    {
        std::string topic{cli->message.topic, cli->message.topic_len};

        std::string payload{cli->message.topic, cli->message.topic_len};

        mqtt_lua_call(mqtt->get_on_message_fn(), topic.c_str(), payload.c_str());
    }
    break;
    case MQTT_TYPE_DISCONNECT:
        spdlog::error("disconnect.");
        break;
    default:
        break;
    }
}

int MQTT::open(const char* client_id,
               const char* username,
               const char* password,
               int         keep_alive,
               const char* host,
               int         port,
               const char* on_connected,
               const char* on_message)
{
    on_connected_ = on_connected;
    on_message_   = on_message;

    std::thread([&]() {
        cli = mqtt_client_new(NULL);
        if (cli == NULL)
        {
            spdlog::error("mqtt client init faild.");
            return;
        }

        cli->keepalive     = keep_alive;
        cli->clean_session = 1;
        cli->userdata      = this;

        mqtt_client_set_id(cli, client_id);
        mqtt_client_set_auth(cli, username, password);

        mqtt_client_set_callback(cli, on_mqtt);

        reconn_setting_t reconn_setting;
        reconn_setting_init(&reconn_setting);
        reconn_setting.min_delay    = 1000;
        reconn_setting.max_delay    = 30000;
        reconn_setting.delay_policy = 2;
        cli->reconn_setting         = &reconn_setting;

        mqtt_client_connect(cli, host, port, 0);

        mqtt_client_run(cli);
    }).detach();

    if (!latch_.wait(5000))
    {
        spdlog::error("MQTT connect timeout");

        mqtt_client_stop(cli);
        return -1;
    }
    return 0;
}

void MQTT::subscribe(const char* topic, int qos)
{
    mqtt_client_subscribe(cli, topic, qos);
}

int MQTT::publish(const char* topic, const char* payload, int qos)
{
    mqtt_message_t msg{0};
    msg.topic       = topic;
    msg.topic_len   = strlen(topic);
    msg.payload     = payload;
    msg.payload_len = strlen(payload);
    msg.qos         = qos;
    return mqtt_client_publish(cli, &msg);
}

MQTT* newMQTT()
{
    auto p = new MQTT;
    return p;
}
